package coordination

import (
	"context"
	"encoding/json"
	"fmt"
	"log"
	"os"
	"sync"
	"time"

	"github.com/louloulin/dataflare/pkg/cluster"
)

// DistributedLock 是分布式锁接口
type DistributedLock interface {
	// Lock 获取锁
	Lock(ctx context.Context) error
	// Unlock 释放锁
	Unlock(ctx context.Context) error
	// TryLock 尝试获取锁
	TryLock(ctx context.Context) (bool, error)
	// IsLocked 检查锁是否被获取
	IsLocked(ctx context.Context) (bool, error)
	// IsOwner 检查是否为锁的拥有者
	IsOwner(ctx context.Context) (bool, error)
	// GetOwner 获取锁的拥有者
	GetOwner(ctx context.Context) (*cluster.NodeInfo, error)
}

// ClusterLock 是基于集群的分布式锁
type ClusterLock struct {
	// manager 是集群管理器
	manager cluster.Manager
	// lockKey 是锁键
	lockKey string
	// ownerID 是锁拥有者ID
	ownerID string
	// isOwner 表示本地节点是否为锁拥有者
	isOwner bool
	// mu 是保护状态的互斥锁
	mu sync.RWMutex
	// logger 是日志记录器
	logger *log.Logger
	// ctx 是上下文
	ctx context.Context
	// cancel 是取消函数
	cancel context.CancelFunc
	// renewalTicker 是续约定时器
	renewalTicker *time.Ticker
	// lockTimeout 是锁超时时间
	lockTimeout time.Duration
	// renewalInterval 是续约间隔
	renewalInterval time.Duration
}

// NewClusterLock 创建一个新的基于集群的分布式锁
func NewClusterLock(manager cluster.Manager, lockKey string, lockTimeout time.Duration) DistributedLock {
	// 创建上下文
	ctx, cancel := context.WithCancel(context.Background())

	// 计算续约间隔（锁超时的1/3）
	renewalInterval := lockTimeout / 3
	if renewalInterval < time.Second {
		renewalInterval = time.Second
	}

	return &ClusterLock{
		manager:         manager,
		lockKey:         lockKey,
		lockTimeout:     lockTimeout,
		renewalInterval: renewalInterval,
		logger:          log.New(os.Stderr, "[DistributedLock] ", log.LstdFlags),
		ctx:             ctx,
		cancel:          cancel,
	}
}

// Lock 获取锁
func (l *ClusterLock) Lock(ctx context.Context) error {
	// 尝试获取锁
	for {
		// 检查上下文是否取消
		select {
		case <-ctx.Done():
			return ctx.Err()
		default:
		}

		// 尝试获取锁
		acquired, err := l.TryLock(ctx)
		if err != nil {
			return err
		}

		// 如果获取成功，返回
		if acquired {
			return nil
		}

		// 等待一段时间后重试
		select {
		case <-ctx.Done():
			return ctx.Err()
		case <-time.After(100 * time.Millisecond):
		}
	}
}

// Unlock 释放锁
func (l *ClusterLock) Unlock(ctx context.Context) error {
	l.mu.Lock()
	defer l.mu.Unlock()

	// 检查是否为锁的拥有者
	if !l.isOwner {
		return fmt.Errorf("not the lock owner")
	}

	// 停止续约
	if l.renewalTicker != nil {
		l.renewalTicker.Stop()
		l.renewalTicker = nil
	}

	// 删除锁元数据
	err := l.manager.SetMetadata(ctx, l.getLockMetadataKey(), "")
	if err != nil {
		return fmt.Errorf("failed to release lock: %w", err)
	}

	// 更新状态
	l.isOwner = false
	l.ownerID = ""

	l.logger.Printf("Released lock: %s", l.lockKey)
	return nil
}

// TryLock 尝试获取锁
func (l *ClusterLock) TryLock(ctx context.Context) (bool, error) {
	l.mu.Lock()
	defer l.mu.Unlock()

	// 检查是否已经拥有锁
	if l.isOwner {
		return true, nil
	}

	// 获取本地节点
	localNode, err := l.manager.GetLocalNode(ctx)
	if err != nil {
		return false, fmt.Errorf("failed to get local node: %w", err)
	}

	// 获取当前锁拥有者
	owner, err := l.getOwnerInternal(ctx)
	if err == nil && owner != nil {
		// 检查锁是否过期
		lockValue, err := l.manager.GetMetadata(ctx, owner.ID, l.getLockMetadataKey())
		if err == nil && lockValue != "" {
			// 解析锁值
			var lockInfo struct {
				Timestamp int64 `json:"timestamp"`
			}
			if err := json.Unmarshal([]byte(lockValue), &lockInfo); err == nil {
				// 检查锁是否过期
				if time.Since(time.Unix(lockInfo.Timestamp, 0)) < l.lockTimeout {
					// 锁未过期
					l.ownerID = owner.ID
					return false, nil
				}
			}
		}
	}

	// 创建锁值
	lockInfo := struct {
		Timestamp int64 `json:"timestamp"`
	}{
		Timestamp: time.Now().Unix(),
	}
	lockValue, err := json.Marshal(lockInfo)
	if err != nil {
		return false, fmt.Errorf("failed to marshal lock info: %w", err)
	}

	// 尝试获取锁
	err = l.manager.SetMetadata(ctx, l.getLockMetadataKey(), string(lockValue))
	if err != nil {
		return false, fmt.Errorf("failed to acquire lock: %w", err)
	}

	// 等待一段时间，确保锁已经在集群中传播
	time.Sleep(100 * time.Millisecond)

	// 检查是否成功获取锁
	owner, err = l.getOwnerInternal(ctx)
	if err != nil {
		return false, fmt.Errorf("failed to get lock owner: %w", err)
	}

	// 如果锁的拥有者是本地节点，则获取成功
	if owner != nil && owner.ID == localNode.ID {
		// 更新状态
		l.isOwner = true
		l.ownerID = owner.ID

		// 启动续约
		l.startRenewal(ctx)

		l.logger.Printf("Acquired lock: %s", l.lockKey)
		return true, nil
	}

	return false, nil
}

// IsLocked 检查锁是否被获取
func (l *ClusterLock) IsLocked(ctx context.Context) (bool, error) {
	owner, err := l.GetOwner(ctx)
	if err != nil {
		return false, err
	}

	return owner != nil, nil
}

// IsOwner 检查是否为锁的拥有者
func (l *ClusterLock) IsOwner(ctx context.Context) (bool, error) {
	l.mu.RLock()
	defer l.mu.RUnlock()

	return l.isOwner, nil
}

// GetOwner 获取锁的拥有者
func (l *ClusterLock) GetOwner(ctx context.Context) (*cluster.NodeInfo, error) {
	l.mu.RLock()
	defer l.mu.RUnlock()

	return l.getOwnerInternal(ctx)
}

// getOwnerInternal 获取锁的拥有者（内部方法，不加锁）
func (l *ClusterLock) getOwnerInternal(ctx context.Context) (*cluster.NodeInfo, error) {
	// 获取所有节点
	nodes, err := l.manager.GetNodes(ctx)
	if err != nil {
		return nil, fmt.Errorf("failed to get nodes: %w", err)
	}

	// 查找锁的拥有者
	for _, node := range nodes {
		if node.Status == cluster.NodeStatusUp {
			// 获取锁元数据
			lockValue, err := l.manager.GetMetadata(ctx, node.ID, l.getLockMetadataKey())
			if err == nil && lockValue != "" {
				// 解析锁值
				var lockInfo struct {
					Timestamp int64 `json:"timestamp"`
				}
				if err := json.Unmarshal([]byte(lockValue), &lockInfo); err == nil {
					// 检查锁是否过期
					if time.Since(time.Unix(lockInfo.Timestamp, 0)) < l.lockTimeout {
						// 找到锁的拥有者
						return node, nil
					}
				}
			}
		}
	}

	return nil, nil
}

// getLockMetadataKey 获取锁元数据键
func (l *ClusterLock) getLockMetadataKey() string {
	return fmt.Sprintf("lock:%s", l.lockKey)
}

// startRenewal 启动锁续约
func (l *ClusterLock) startRenewal(ctx context.Context) {
	// 停止旧的续约
	if l.renewalTicker != nil {
		l.renewalTicker.Stop()
	}

	// 创建新的续约定时器
	l.renewalTicker = time.NewTicker(l.renewalInterval)

	// 启动续约协程
	go func() {
		for {
			select {
			case <-l.ctx.Done():
				return
			case <-ctx.Done():
				return
			case <-l.renewalTicker.C:
				// 续约锁
				l.renewLock(ctx)
			}
		}
	}()
}

// renewLock 续约锁
func (l *ClusterLock) renewLock(ctx context.Context) {
	l.mu.Lock()
	defer l.mu.Unlock()

	// 检查是否为锁的拥有者
	if !l.isOwner {
		return
	}

	// 创建锁值
	lockInfo := struct {
		Timestamp int64 `json:"timestamp"`
	}{
		Timestamp: time.Now().Unix(),
	}
	lockValue, err := json.Marshal(lockInfo)
	if err != nil {
		l.logger.Printf("Failed to marshal lock info: %v", err)
		return
	}

	// 更新锁元数据
	err = l.manager.SetMetadata(ctx, l.getLockMetadataKey(), string(lockValue))
	if err != nil {
		l.logger.Printf("Failed to renew lock: %v", err)
		return
	}

	l.logger.Printf("Renewed lock: %s", l.lockKey)
}
